﻿using CatClient.Common;
using CatClient.Configuration;
using CatClient.Fusing;
using CatClient.Message.Internals;
using CatClient.Message.Spi.Codec;
using CatClient.Util;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Net.Sockets;
using System.Threading;

namespace CatClient.Message.Spi.IO
{
    public class TcpMessageSender : IMessageSender
    {
        private readonly ClientConfig _clientConfig;

        private readonly IMessageCodec _codec;

        private readonly ConcurrentQueue<IMessageTree> _queue;

        private readonly ConcurrentQueue<IMessageTree> _atomicTress;

        private readonly ConcurrentList<ServerSocket> _connPool;

        private readonly IMessageStatistics _statistics;

        private volatile int _activeServerIndex;

        private readonly object _serverLocker = new object();

        private long _errors;

        private bool _active;

        private readonly int _maxQueueSize;

        private long _sendMessageNullCount;

        private long MaxSizeOfMessage
        {
            get
            {
                return CatConfigManager.GetConfigObject<long>("MsgMaxSize", 500L);
            }
        }

        public virtual bool HasSendingMessage
        {
            get
            {
                return this._queue.Count > 0 && this._connPool.Count > 0;
            }
        }

        public TcpMessageSender(ClientConfig clientConfig, IMessageStatistics statistics)
        {
            this._clientConfig = clientConfig;
            this._statistics = statistics;
            this._connPool = new ConcurrentList<ServerSocket>();
            this._queue = new ConcurrentQueue<IMessageTree>();
            this._atomicTress = new ConcurrentQueue<IMessageTree>();
            this._codec = new PlainTextMessageCodec();
            this._maxQueueSize = clientConfig.Domain.MaxQueueSize;
        }

        public void Initialize()
        {
            ThreadPool.QueueUserWorkItem(new WaitCallback(this.ServerManagementTask));
            LogInfoWriter.GetInstance("Cat", 0, LogInfoWriter.LogLayout.Default).Info("Thread(ServerManagementTask) started.");
            if (this._clientConfig.Domain.ThreadPool != 1)
            {
                this._clientConfig.Domain.ThreadPool = 1;
                LogInfoWriter.GetInstance("Cat", 0, LogInfoWriter.LogLayout.Default).Warn("暂时禁止多线程发送消息，系统自动将发送线程数修改为1");
            }
            for (int i = 0; i < this._clientConfig.Domain.ThreadPool; i++)
            {
                ThreadPool.QueueUserWorkItem(new WaitCallback(this.AsynchronousSendTask), i);
                LogInfoWriter.GetInstance("Cat", 0, LogInfoWriter.LogLayout.Default).Info(string.Format("Thread(AsynchronousSendTask-{0}) started.", i));
            }
            ThreadPool.QueueUserWorkItem(new WaitCallback(this.MergeAtomicTask));
            LogInfoWriter.GetInstance("Cat", 0, LogInfoWriter.LogLayout.Default).Info("Thread(MergeAtomicTask) started.");
            this._active = true;
        }

        public void Send(IMessageTree tree)
        {
            if (tree == null)
                return;
            
            if (tree.Message == null)
            {
                bool flag = true;
                if (!CatConfigManager.GetConfigObject<bool>("CatNullMsgLog", false))
                {
                    flag = (Interlocked.Increment(ref this._sendMessageNullCount) % (long)CatConfigManager.GetConfigObject<int>("DefaultLogCount", 1000) == 0L);
                }
                if (flag)
                {
                    try
                    {
                        LogInfoWriter.GetInstance("Cat", 0, LogInfoWriter.LogLayout.Default).Error("CatNullMsg Message Null: " + Environment.StackTrace);
                    }
                    catch (Exception exception)
                    {
                        LogInfoWriter.GetInstance("Cat", 0, LogInfoWriter.LogLayout.Default).Error("CatNullMsg Exception", exception);
                    }
                }
                return;
            }
            if (this.IsAtomicMessage(tree))
            {
                if (this._atomicTress.Count < this._maxQueueSize)
                {
                    this._atomicTress.Enqueue(tree);
                    return;
                }
                this.LogQueueFullInfo("AtomicMessage");
                return;
            }
            else
            {
                if (this._queue.Count < this._maxQueueSize)
                {
                    this._queue.Enqueue(tree);
                    return;
                }
                this.LogQueueFullInfo("Message");
                return;
            }
        }

        public void Shutdown()
        {
            try
            {
                this._active = false;
            }
            catch
            {
            }
        }

        public void ServerManagementTask(object o)
        {
            while (true)
            {
                try
                {
                    if (this._active)
                    {
                        LogInfoWriter.GetInstance("Cat", 0, LogInfoWriter.LogLayout.Default).Info("ServerManagementTask config init start.");
                        Stopwatch stopwatch = new Stopwatch();
                        stopwatch.Start();
                        this._clientConfig.Initialize();
                        stopwatch.Stop();
                        LogInfoWriter.GetInstance("Cat", 0, LogInfoWriter.LogLayout.Default).Info(string.Format("ServerManagementTask end. {0}ms", stopwatch.ElapsedMilliseconds));
                        LogInfoWriter.GetInstance("Cat", 0, LogInfoWriter.LogLayout.Default).Info("ServerManagementTask connect check start.");
                        stopwatch.Restart();
                        this.ReloadConnectionPool();
                        stopwatch.Stop();
                        LogInfoWriter.GetInstance("Cat", 0, LogInfoWriter.LogLayout.Default).Info(string.Format("ServerManagementTask connect check end. {0}ms", stopwatch.ElapsedMilliseconds));
                    }
                    MilliSecondTimer.SleepToNextMinute(20000);
                }
                catch (Exception exception)
                {
                    LogInfoWriter.GetInstance("Cat", 0, LogInfoWriter.LogLayout.Default).Error("Error when pull server config", exception.ToCatException());
                    Thread.Sleep(5000);
                }
            }
        }

        private void ReloadConnectionPool()
        {
            object serverLocker = this._serverLocker;
            lock (serverLocker)
            {
                this._activeServerIndex = -1;
                int num = 0;
                using (List<Server>.Enumerator enumerator = this._clientConfig.Servers.GetEnumerator())
                {
                    while (enumerator.MoveNext())
                    {
                        Server server = enumerator.Current;
                        ServerSocket serverSocket = this._connPool.FirstOrDefault((ServerSocket t) => t.Server.Equals(server));
                        if (serverSocket != null)
                        {
                            this._connPool.Remove(serverSocket);
                            if (serverSocket.Channel == null || !serverSocket.Channel.Connected)
                            {
                                serverSocket.Channel = this.CreateChannel(serverSocket.Server);
                            }
                        }
                        else
                        {
                            TcpClient channel = this.CreateChannel(server);
                            serverSocket = new ServerSocket
                            {
                                Server = server,
                                Channel = channel
                            };
                        }
                        if (serverSocket.Channel != null && serverSocket.Channel.Connected)
                        {
                            this._connPool.Insert(num, serverSocket);
                            num++;
                        }
                        this._activeServerIndex = (this._activeServerIndex + 1) % this._clientConfig.Servers.Count;
                        if (num >= this._clientConfig.Domain.ThreadPool)
                        {
                            for (int i = num; i < this._connPool.Count; i++)
                            {
                                ServerSocket serverSocket2 = this._connPool[i];
                                if (serverSocket2.Channel != null && serverSocket2.Channel.Connected)
                                {
                                    serverSocket2.Channel.Close();
                                }
                                this._connPool.Remove(serverSocket2);
                            }
                            break;
                        }
                    }
                }
            }
        }

        public void AsynchronousSendTask(object o)
        {
            int num = 0;
            IMessageTree messageTree = null;
            while (true)
            {
                if (!this._active || !CatConfig.Config.Domain.Enabled)
                {
                    Thread.Sleep(10000);
                }
                else
                {
                    try
                    {
                        ServerSocket serverSocket = this.GetServerSocket((int)o);
                        if (serverSocket == null)
                        {
                            Thread.Sleep(5000);
                        }
                        else
                        {
                            if (messageTree == null)
                            {
                                num = 0;
                                this._queue.TryDequeue(out messageTree);
                                if (messageTree == null)
                                {
                                    Thread.Sleep(500);
                                    continue;
                                }
                            }
                            this.SendInternal(messageTree, serverSocket.Channel);
                            messageTree.Message = null;
                            messageTree = null;
                        }
                    }
                    catch (Exception exception)
                    {
                        LogInfoWriter.GetInstance("Cat", 0, LogInfoWriter.LogLayout.Default).Error("Error when sending message over TCP socket! Error: {0}", exception.ToCatException());
                        if (++num > 10)
                        {
                            try
                            {
                                Cat.GetFuseManager().Increment(FuseType.Connection);
                                LogInfoWriter.GetInstance("Cat", 0, LogInfoWriter.LogLayout.Default).Warn("Message send error, had try 10 times but failed. message info: " + messageTree);
                            }
                            catch (Exception exception2)
                            {
                                LogInfoWriter.GetInstance("Cat", 0, LogInfoWriter.LogLayout.Default).Error("CatNullMsg catch null", exception2);
                            }
                            finally
                            {
                                messageTree = null;
                            }
                        }
                        Thread.Sleep(1000);
                    }
                }
            }
        }

        private ServerSocket GetServerSocket(int idx)
        {
            object serverLocker = this._serverLocker;
            ServerSocket result;
            lock (serverLocker)
            {
                if (this._clientConfig.Servers.Count == 0)
                {
                    result = null;
                }
                else if (this._connPool.Count == 0)
                {
                    int activeServerIndex = this._activeServerIndex;
                    Server server;
                    TcpClient tcpClient;
                    while (true)
                    {
                        this._activeServerIndex = (this._activeServerIndex + 1) % this._clientConfig.Servers.Count;
                        server = this._clientConfig.Servers[this._activeServerIndex];
                        tcpClient = this.CreateChannel(server);
                        if (tcpClient != null && tcpClient.Connected)
                        {
                            break;
                        }
                        if (activeServerIndex == this._activeServerIndex)
                        {
                            goto Block_7;
                        }
                    }
                    ServerSocket serverSocket = new ServerSocket
                    {
                        Server = server,
                        Channel = tcpClient
                    };
                    this._connPool.Add(serverSocket);
                    result = serverSocket;
                    return result;
                    Block_7:
                    result = null;
                }
                else if (this._connPool.Count == 0)
                {
                    result = null;
                }
                else
                {
                    int index = idx % this._connPool.Count;
                    ServerSocket serverSocket2 = this._connPool[index];
                    if (serverSocket2.Channel == null || !serverSocket2.Channel.Connected)
                    {
                        serverSocket2.Channel = this.CreateChannel(serverSocket2.Server);
                        if (serverSocket2.Channel == null || !serverSocket2.Channel.Connected)
                        {
                            this._connPool.Remove(serverSocket2);
                            result = null;
                            return result;
                        }
                    }
                    result = serverSocket2;
                }
            }
            return result;
        }

        public void MergeAtomicTask(object o)
        {
            while (true)
            {
                if (!CatConfig.Config.Domain.Enabled)
                {
                    Thread.Sleep(10000);
                }
                else
                {
                    try
                    {
                        if (this.ShouldMerge())
                        {
                            IMessageTree messageTree = this.MergeTree();
                            if (messageTree == null)
                            {
                                Thread.Sleep(5000);
                            }
                            else
                            {
                                this.Send(messageTree);
                            }
                        }
                        else
                        {
                            Thread.Sleep(100);
                        }
                    }
                    catch (Exception exception)
                    {
                        LogInfoWriter.GetInstance("Cat", 0, LogInfoWriter.LogLayout.Default).Error("Error when merge tree message", exception.ToCatException());
                        Thread.Sleep(5000);
                    }
                }
            }
        }

        private void SendInternal(IMessageTree tree, TcpClient activeChannel)
        {
            if (activeChannel == null || !activeChannel.Connected)
            {
                LogInfoWriter.GetInstance("Cat", 0, LogInfoWriter.LogLayout.Default).Warn("SendInternal中，Socket关闭");
                return;
            }
            ChannelBuffer channelBuffer = new ChannelBuffer(8192);
            this._codec.Encode(tree, channelBuffer);
            byte[] array = channelBuffer.ToArray();
            if ((long)array.Length / 1024L > this.MaxSizeOfMessage)
            {
                LogInfoWriter.GetInstance("Cat", 0, LogInfoWriter.LogLayout.Default).Error(string.Format("Large message: {0}{1}{2}", Environment.NewLine, tree, Environment.NewLine));
                return;
            }
            activeChannel.Client.Send(array);
            IMessageStatistics expr_79 = this._statistics;
            if (expr_79 == null)
            {
                return;
            }
            expr_79.OnBytes(array.Length);
        }

        private TcpClient CreateChannel(Server server)
        {
            if (!server.Enabled)
            {
                return null;
            }
            TcpClient tcpClient = new TcpClient
            {
                NoDelay = true,
                ReceiveTimeout = 2000
            };
            string ip = server.Ip;
            int port = server.Port;
            LogInfoWriter.GetInstance("Cat", 0, LogInfoWriter.LogLayout.Default).Info(string.Format("Connecting to server({0}:{1}) ...", ip, port));
            try
            {
                tcpClient.Connect(ip, port);
                if (tcpClient.Connected)
                {
                    LogInfoWriter.GetInstance("Cat", 0, LogInfoWriter.LogLayout.Default).Info(string.Format("Connected to server({0}:{1}).", ip, port));
                    return tcpClient;
                }
                LogInfoWriter.GetInstance("Cat", 0, LogInfoWriter.LogLayout.Default).Error(string.Format("Failed to connect to server({0}:{1}).", ip, port));
            }
            catch (Exception ex)
            {
                LogInfoWriter.GetInstance("Cat", 0, LogInfoWriter.LogLayout.Default).Error(string.Format("Failed to connect to server({0}:{1}). Error: {2}.", ip, port, ex.Message));
            }
            return null;
        }

        private bool ShouldMerge()
        {
            IMessageTree messageTree = null;
            if (!this._atomicTress.TryDequeue(out messageTree))
            {
                return false;
            }
            long timestamp = messageTree.Message.Timestamp;
            int num = 30000;
            return MilliSecondTimer.CurrentTimeMillis - timestamp > (long)num || this._atomicTress.Count >= 200;
        }

        private IMessageTree MergeTree()
        {
            int num = 200;
            DefaultTransaction defaultTransaction = new DefaultTransaction("_CatMergeTree", "_CatMergeTree", null);
            IMessageTree messageTree = null;
            if (!this._atomicTress.TryDequeue(out messageTree))
            {
                return null;
            }
            defaultTransaction.Status = "0";
            defaultTransaction.Complete();
            defaultTransaction.AddChild(messageTree.Message);
            defaultTransaction.Timestamp = messageTree.Message.Timestamp;
            long num2 = 0L;
            long num3 = 0L;
            while (num-- >= 0)
            {
                IMessageTree messageTree2 = null;
                if (!this._atomicTress.TryDequeue(out messageTree2))
                {
                    defaultTransaction.DurationInMillis = num2 - defaultTransaction.Timestamp + num3;
                    break;
                }
                num2 = messageTree2.Message.Timestamp;
                if (messageTree2.Message is DefaultTransaction)
                {
                    num3 = ((DefaultTransaction)messageTree2.Message).DurationInMillis;
                }
                else
                {
                    num3 = 0L;
                }
                defaultTransaction.AddChild(messageTree2.Message);
            }
            messageTree.Message = defaultTransaction;
            return messageTree;
        }

        private bool IsAtomicMessage(IMessageTree tree)
        {
            IMessage message = tree.Message;
            if (message is ITransaction)
            {
                string type = message.Type;
                return type.StartsWith("Cache.") || type.StartsWith("SQL");
            }
            return true;
        }

        private void LogQueueFullInfo(string name)
        {
            Interlocked.Increment(ref this._errors);
            IMessageStatistics expr_12 = this._statistics;
            if (expr_12 != null)
            {
                expr_12.OnOverflowed();
            }
            if (Interlocked.Read(ref this._errors) % 100L == 0L)
            {
                LogInfoWriter.GetInstance("Cat", 0, LogInfoWriter.LogLayout.Default).Warn(string.Format("{0} queue's full! Count: {1}", name, Interlocked.Read(ref this._errors)));
            }
        }

        public void Clear()
        {
            IMessageTree messageTree;
            while (this._queue.TryDequeue(out messageTree))
            {
            }
            while (this._atomicTress.TryDequeue(out messageTree))
            {
            }
        }
    }
}
